03 使用Python开发MCP Server
Python MCP SDK提供了FastMCP封装,可以通过装饰器快速定义Tools、Resources和Prompts。对于大多数业务场景,可以先用FastMCP完成Server原型,再根据安全、部署和权限需求逐步增强。
本文实现一个文档查询MCP Server,包含:
- 一个搜索工具:
search_docs - 一个文档资源:
docs://article/{slug} - 一个提示词模板:
summarize_article
一、项目初始化
创建项目目录:
uv init docs-mcp-server
cd docs-mcp-server
uv venv
source .venv/bin/activate安装依赖:
uv add "mcp[cli]"创建入口文件:
touch server.py项目结构如下:
docs-mcp-server/
├── server.py
└── pyproject.toml二、创建FastMCP实例
在server.py中初始化MCP Server:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("docs-server")FastMCP("docs-server")中的名称用于标识当前Server。Host或调试工具连接Server时,可以通过该名称区分不同服务。
三、准备示例数据
为了让示例可以独立运行,先用内存字典模拟文档库:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("docs-server")
ARTICLES = {
"docker-intro": {
"title": "Docker基础入门",
"content": "Docker用于把应用和运行环境打包成镜像,减少环境差异。",
"tags": ["docker", "container"],
},
"mcp-architecture": {
"title": "MCP基础与架构",
"content": "MCP通过Host、Client、Server连接AI应用和外部系统。",
"tags": ["mcp", "agent"],
},
}真实项目中,ARTICLES可以替换成数据库查询、向量检索、文件读取或内部HTTP API。
四、暴露Tool
工具用于执行动作。这里定义一个search_docs工具,用于按关键词搜索文章:
@mcp.tool()
def search_docs(query: str) -> list[dict]:
"""Search articles by keyword.
Args:
query: Keyword used to search article title, content, or tags.
"""
keyword = query.lower()
results = []
for slug, article in ARTICLES.items():
haystack = " ".join(
[article["title"], article["content"], *article["tags"]]
).lower()
if keyword in haystack:
results.append(
{
"slug": slug,
"title": article["title"],
"summary": article["content"],
}
)
return resultsFastMCP会根据函数签名和docstring生成工具定义,包括工具名称、描述和参数schema。模型能否正确使用工具,很大程度取决于函数命名、参数类型和docstring是否清晰。
五、暴露Resource
Resource用于提供只读上下文。这里定义一个按slug读取文章的资源:
@mcp.resource("docs://article/{slug}")
def read_article(slug: str) -> str:
"""Read an article by slug."""
article = ARTICLES.get(slug)
if not article:
raise ValueError(f"Article not found: {slug}")
return f"# {article['title']}\n\n{article['content']}"资源URI是Host识别和读取资源的入口。docs://article/{slug}表示这是一个带参数的资源模板。
Resource适合承载上下文数据,例如文档正文、数据库schema、配置说明等。与Tool不同,Resource通常不应该修改外部系统状态。
六、暴露Prompt
Prompt用于提供可复用的任务模板。这里定义一个文章总结模板:
@mcp.prompt()
def summarize_article(slug: str) -> str:
"""Create a prompt for summarizing an article."""
return (
"请读取资源 docs://article/{slug},并按以下结构总结:\n"
"1. 核心主题\n"
"2. 关键概念\n"
"3. 适用场景\n"
"4. 注意事项\n"
).format(slug=slug)Prompt适合沉淀标准任务流程。例如代码审查、故障排查、文章总结、需求分析等。
七、启动Server
本地MCP Server通常使用stdio传输。添加启动入口:
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()完整文件如下:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("docs-server")
ARTICLES = {
"docker-intro": {
"title": "Docker基础入门",
"content": "Docker用于把应用和运行环境打包成镜像,减少环境差异。",
"tags": ["docker", "container"],
},
"mcp-architecture": {
"title": "MCP基础与架构",
"content": "MCP通过Host、Client、Server连接AI应用和外部系统。",
"tags": ["mcp", "agent"],
},
}
@mcp.tool()
def search_docs(query: str) -> list[dict]:
"""Search articles by keyword.
Args:
query: Keyword used to search article title, content, or tags.
"""
keyword = query.lower()
results = []
for slug, article in ARTICLES.items():
haystack = " ".join(
[article["title"], article["content"], *article["tags"]]
).lower()
if keyword in haystack:
results.append(
{
"slug": slug,
"title": article["title"],
"summary": article["content"],
}
)
return results
@mcp.resource("docs://article/{slug}")
def read_article(slug: str) -> str:
"""Read an article by slug."""
article = ARTICLES.get(slug)
if not article:
raise ValueError(f"Article not found: {slug}")
return f"# {article['title']}\n\n{article['content']}"
@mcp.prompt()
def summarize_article(slug: str) -> str:
"""Create a prompt for summarizing an article."""
return (
"请读取资源 docs://article/{slug},并按以下结构总结:\n"
"1. 核心主题\n"
"2. 关键概念\n"
"3. 适用场景\n"
"4. 注意事项\n"
).format(slug=slug)
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()启动Server:
uv run server.pystdio模式下,Server会等待MCP Client通过标准输入输出发送JSON-RPC消息。直接运行时终端可能不会出现普通Web服务那样的监听地址。
八、stdio日志注意事项
使用stdio传输时,Server的stdout用于发送MCP协议消息。因此,业务日志不能写到stdout,否则会破坏JSON-RPC通信。
错误示例:
print("server started")推荐写法:
import sys
print("server started", file=sys.stderr)也可以使用日志库,并确保日志输出到stderr或文件。
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
logger.info("server started")对于HTTP传输,标准输出不会干扰HTTP响应,但仍建议使用结构化日志,便于排查工具调用和权限问题。
九、开发建议
开发MCP Server时,可以按以下顺序推进:
- 先用内存数据或本地文件完成最小Server。
- 用
@mcp.tool()暴露最核心的查询类工具。 - 再补充Resource,让Host可以读取上下文。
- 将重复任务沉淀为Prompt。
- 使用MCP Inspector测试工具列表、资源读取和提示词模板。
- 接入真实系统前补充参数校验、权限控制和日志。
十、总结
使用Python开发MCP Server的关键步骤如下:
| 步骤 | 说明 |
|---|---|
| 初始化Server | 创建FastMCP实例 |
| 暴露Tool | 使用@mcp.tool()定义可执行函数 |
| 暴露Resource | 使用@mcp.resource()定义可读取上下文 |
| 暴露Prompt | 使用@mcp.prompt()定义提示词模板 |
| 启动Server | 本地开发通常使用stdio传输 |
| 调试 | 使用Inspector检查能力发现和调用结果 |
完成最小Server后,就可以进入传输方式、客户端接入和部署配置。